#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：henan_sumo
@File ：main_send.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/4/16 17:48
'''
import threading
from multiprocessing import Queue, Process

from kafka.errors import KafkaError

from config import KAFKA_HOST, KAFKA_E1_TOPIC, TYPE_CHANGE, KAFKA_E1_FUSION_TOPIC
from loguru import logger
import time
import json
from kafka import KafkaConsumer, KafkaProducer
from utils_t import time_str

sn_dict = {}
for num in range(17, 39):
    sn_dict[num] = None


# sn_dict[12] = None

def read_e1_channel(time_data):
    # try:
    # 实时接收卡口数据
    cache_time = None
    kafka_consumer_e1 = KafkaConsumer(
        KAFKA_E1_TOPIC,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                                   value_serializer=lambda x: json.dumps(x).encode('utf-8'))
    logger.info(f"事件{KAFKA_E1_TOPIC}监听已就绪")
    first_frame = int(time.time())
    logger.info(f"{str(sn_dict.keys())}")
    while True:
        time.sleep(0.1)
        target_data = []
        target_result = {}
        for message in kafka_consumer_e1:
            if message:
                try:
                    # 获取消息时间戳

                    msg_timestamp = message.timestamp  # 注意这里根据实际情况可能需要调整
                    # logger.info(f"消息时间戳: {msg_timestamp}")
                    message_value = message.value.decode('utf-8')
                    E1_info = json.loads(message_value)
                    # targets = E1_info['targets']
                    if E1_info['deviceId'] == 'hawkeye_server':
                        continue
                    else:
                        try:
                            deviceId = int(E1_info['deviceId'].split(",")[0])
                        except Exception as e:
                            logger.error(str(e))
                            continue
                    if deviceId in sn_dict.keys():
                        sn_dict[deviceId] = E1_info
                    else:
                        continue

                    temp_list = []

                    for sn, info in sn_dict.items():
                        if info is None:
                            continue
                        targets = info['targets']
                        for one in targets:
                            # print(E1_info['deviceId'])
                            heading = one['heading']
                            lane = one['lane']
                            latitude = one['latitude']
                            longitude = one['longitude']
                            speed = one['speed']
                            car_type = one['type']
                            uuid = one['uuid']
                            length = one['len'] if 'len' in one.keys() else 0
                            height = one['height'] if 'height' in one.keys() else 0
                            weight = one['weight'] if 'weight' in one.keys() else 0
                            car_license = one['license']
                            vehicleColor = one['vehicleColor']
                            vehicleType = one['vehicleType']
                            temp = {"id": uuid, "type": int(car_type), 'source': 2, "laneNum": lane,
                                    "longitude": longitude, "latitude": latitude, "speed": speed,
                                    "courseAngle": heading,
                                    "length": length, "width": weight, "height": height,
                                    "sourceId": sn,
                                    "plate": car_license, "color": vehicleColor, "vehicleType": vehicleType
                                    }
                            temp_list.append(temp)
                    # now_timestamp = E1_info['timestamp']
                    if cache_time is not None:
                        logger.info(f"{cache_time}")
                        if (int(msg_timestamp / 100) - cache_time) >= 1:
                            # logger.info(f"发送一次数据{target_data}")
                            target_result['globalTimeStamp'] = time_str(cache_time * 100)
                            target_result['participantNum'] = len(temp_list)
                            target_result['globalFrameNo'] = first_frame
                            target_result['e1FrameParticipant'] = temp_list
                            first_frame += 1
                            logger.info(f"发送一帧时间戳{msg_timestamp}")
                            time_data.put(int(time.time()))
                            try:
                                kafka_producer.send(KAFKA_E1_FUSION_TOPIC, value=target_result,
                                                             key=str(cache_time))
                            except KafkaError as e:
                                logger.error(f"kafka断开连接")
                            cache_time = int(msg_timestamp / 100)
                    else:
                        cache_time = int(msg_timestamp / 100)
                except Exception as e:
                    print(str(e))


if __name__ == '__main__':
    time_data = Queue(maxsize=2000)  # 创建队列
    tess_process = Process(target=read_e1_channel, args=(time_data,))
    tess_process.start()
    time_simu = int(time.time())
    while True:
        stop_flag = False
        while time_data.qsize() > 0:
            time_simu = time_data.get()
        if time_simu:
            logger.info(f"时间差{abs(time_simu - int(time.time()))}")
            if abs(time_simu - int(time.time())) > 60:
                stop_flag = True
                time_simu = int(time.time())
        if stop_flag:
            logger.info("关闭开启线程")
            tess_process.terminate()
            tess_process.join()
            logger.info("重新开启线程")
            tess_process = Process(target=read_e1_channel, args=(time_data,))
            tess_process.start()
        time.sleep(1)
